package com.zgh.opcua.manager;

import com.zgh.opcua.entity.ReadNodeEntity;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaMonitoredItem;
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaSubscription;
import org.eclipse.milo.opcua.stack.core.AttributeId;
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
import org.eclipse.milo.opcua.stack.core.types.builtin.DateTime;
import org.eclipse.milo.opcua.stack.core.types.builtin.ExtensionObject;
import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
import org.eclipse.milo.opcua.stack.core.types.builtin.Variant;
import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UInteger;
import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned;
import org.eclipse.milo.opcua.stack.core.types.enumerated.DataChangeTrigger;
import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode;
import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;
import org.eclipse.milo.opcua.stack.core.types.structured.DataChangeFilter;
import org.eclipse.milo.opcua.stack.core.types.structured.MonitoredItemCreateRequest;
import org.eclipse.milo.opcua.stack.core.types.structured.MonitoringParameters;
import org.eclipse.milo.opcua.stack.core.types.structured.ReadValueId;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;

/**
 * @ClassName KepwareSubscriptManager
 * @Description Kepserver订阅管理器
 * @Author Administrator
 * @Date 14:56 2025/11/5
 * @Version 1.0
 */
@Slf4j
public class KepwareSubscriptManager {

    private static final KepwareSubscriptManager INSTANCE = new KepwareSubscriptManager();

    private KepwareSubscriptManager() {
    }

    public static KepwareSubscriptManager getInstance() {
        return INSTANCE;
    }

    /**
     * 批量订阅节点
     */
    public String subscribeNodes(List<ReadNodeEntity> nodes) {
        try {
            OpcUaClient client = OpcUaClientInstance.getInstance().getOpcUaClient();
            if (client == null) {
                return "未连接 OPC UA 客户端";
            }
            if (CollectionUtils.isEmpty(nodes)) {
                return "节点列表为空";
            }
            UaSubscription sub = getOrCreateSubscription();

            int count = createMonitoredItems(sub, nodes);
            return count > 0
                    ? "成功订阅 " + count + " 个新节点"
                    : "没有新节点需要订阅";
        } catch (Exception e) {
            log.error("订阅失败", e);
            return "订阅失败: " + e.getMessage();
        }
    }

    /**
     * 取消订阅节点（根据 NodeId）
     */
    public String unsubscribeNodes(List<ReadNodeEntity> nodes) {
        try {
            OpcUaClient client = OpcUaClientInstance.getInstance().getOpcUaClient();
            List<UaSubscription> subs = client.getSubscriptionManager().getSubscriptions();
            if (CollectionUtils.isEmpty(subs)) {
                return " 当前无订阅对象";
            }

            Set<String> nodeIdentifiers = nodes.stream().map(ReadNodeEntity::getIdentifier).collect(Collectors.toSet());

            UaSubscription sub = subs.get(0);
            List<UaMonitoredItem> toRemove = sub.getMonitoredItems().stream()
                    .filter(item -> {
                        NodeId nodeId = item.getReadValueId().getNodeId();
                        return nodeIdentifiers.contains(String.valueOf(nodeId.getIdentifier()));
                    })
                    .collect(Collectors.toList());

            if (toRemove.isEmpty()) {
                return "无匹配节点可取消";
            }

            sub.deleteMonitoredItems(toRemove).get(3, TimeUnit.SECONDS);
            log.info("已取消订阅节点数: {}", toRemove.size());
            return "已取消 " + toRemove.size() + " 个节点";
        } catch (Exception e) {
            log.error("取消订阅失败", e);
            return "取消订阅失败: " + e.getMessage();
        }
    }

    /**
     * 获取现有订阅对象或新建一个
     */
    private UaSubscription getOrCreateSubscription() throws Exception {
        OpcUaClient client = OpcUaClientInstance.getInstance().getOpcUaClient();
        List<UaSubscription> subs = client.getSubscriptionManager().getSubscriptions();

        if (CollectionUtils.isEmpty(subs)) {
            UaSubscription newSub = client.getSubscriptionManager()
                    .createSubscription(1000.0)
                    .get(3, TimeUnit.SECONDS);
            log.info("创建新的订阅对象");
            return newSub;
        } else {
            log.info("使用已存在的订阅对象");
            return subs.get(0);
        }
    }

    /**
     * 创建监控项（过滤重复节点 + 注册回调）
     */
    private int createMonitoredItems(UaSubscription subscription, List<ReadNodeEntity> nodes) throws Exception {

        // 获取已存在节点
        Set<String> existingKeys = collectExistingKeys(subscription);
        //监控点批量处理
        int totalCreated = 0;
        for (int i = 0; i < nodes.size(); i += 1000) {
            List<ReadNodeEntity> batch = nodes.subList(i, Math.min(i + 1000, nodes.size()));

            // 构建新的监控项请求（过滤重复）
            List<MonitoredItemCreateRequest> requests = buildNewRequests(subscription, batch, existingKeys);

            if (requests.isEmpty()) {
                log.info("没有新的节点需要订阅");
                break;
            }

            // 创建监控项并注册回调
            List<UaMonitoredItem> items = createAndRegisterItems(subscription, requests);
            totalCreated += items.size();

        }


        log.info(" 成功订阅 {} 个新节点", totalCreated);
        return totalCreated;
    }


    /**
     * 获取当前订阅中已存在的节点Key集合，避免重复订阅
     */
    private Set<String> collectExistingKeys(UaSubscription subscription) {
        return subscription.getMonitoredItems().stream()
                .map(item -> item.getReadValueId().getNodeId().getNamespaceIndex() + ":" +
                        item.getReadValueId().getNodeId().getIdentifier())
                .collect(Collectors.toSet());
    }

    /**
     * 构建需要新订阅的监控项请求
     */
    private List<MonitoredItemCreateRequest> buildNewRequests(
            UaSubscription subscription,
            List<ReadNodeEntity> nodes,
            Set<String> existingKeys) {

        List<MonitoredItemCreateRequest> requests = new ArrayList<>();

        for (ReadNodeEntity node : nodes) {
            String key = node.getIndex() + ":" + node.getIdentifier();

            if (existingKeys.contains(key)) {
                log.debug("跳过已订阅节点: {}", key);
                continue;
            }

            try {
                MonitoredItemCreateRequest req = buildMonitorRequest(subscription, node);
                requests.add(req);
                log.debug("添加订阅节点: {}, clientHandle={}", key, node.getClientHandle());
            } catch (Exception e) {
                log.warn("创建节点订阅请求失败: {}", key, e);
            }
        }

        return requests;
    }

    /**
     * 构建单个监控项请求
     *
     * @param subscription 当前订阅对象（UaSubscription），用于生成 clientHandle
     * @param node         要订阅的节点信息（ReadNodeEntity）
     * @return MonitoredItemCreateRequest 监控项请求，可用于创建监控项
     * @throws Exception 异常（如编码过滤器失败、客户端实例不可用等）
     */
    private MonitoredItemCreateRequest buildMonitorRequest(UaSubscription subscription, ReadNodeEntity node) throws Exception {
        // 每个监控项都有一个唯一 clientHandle，用于区分不同监控项的回调
        UInteger clientHandle = node.getClientHandle() != null && node.getClientHandle() > 0
                ? UInteger.valueOf(node.getClientHandle())
                : subscription.nextClientHandle();

        // 同步回写到 node 对象中（方便后续使用）
        node.setClientHandle(Integer.parseInt(clientHandle.toString()));

        // 创建 数据变化过滤器
        DataChangeFilter filter = new DataChangeFilter(DataChangeTrigger.StatusValueTimestamp, Unsigned.uint(0), 0.0);
        OpcUaClient client = OpcUaClientInstance.getInstance().getOpcUaClient();
        ExtensionObject extensionObject = ExtensionObject.encode(client.getSerializationContext(), filter);
        Integer nodeSamplingInterval = node.getSamplingInterval();
        MonitoringParameters params = new MonitoringParameters(
                clientHandle,
                nodeSamplingInterval == null ? 1000.0 : nodeSamplingInterval.doubleValue(),  // 采样周期
                extensionObject,    // 默认过滤器
                Unsigned.uint(10),
                true
        );
        //{目标节点 ID，指定监控哪个属性（此处为 Value），数组索引范围，数据编码}
        ReadValueId readValueId = new ReadValueId(
                new NodeId(node.getIndex(), node.getIdentifier()),
                AttributeId.Value.uid(),
                null,
                null
        );

        return new MonitoredItemCreateRequest(readValueId, MonitoringMode.Reporting, params);
    }

    /**
     * 执行监控项创建并注册回调
     */
    private List<UaMonitoredItem> createAndRegisterItems(
            UaSubscription subscription,
            List<MonitoredItemCreateRequest> requests) throws Exception {

        BiConsumer<UaMonitoredItem, Integer> callback = (item, id) -> {
            registerItemCallback(item);
            log.debug("监控项创建成功: {}, clientHandle={}",
                    item.getReadValueId().getNodeId(), item.getClientHandle());
        };

        // 添加超时控制，防止阻塞
        return subscription.createMonitoredItems(
                TimestampsToReturn.Both,
                requests,
                callback
        ).get(5, TimeUnit.SECONDS);
    }

    /**
     * 注册监控项值变化回调
     */
    private void registerItemCallback(UaMonitoredItem item) {
        item.setValueConsumer(this::onValueChanged);
    }

    private void onValueChanged(UaMonitoredItem monitoredItem, DataValue dataValue) {

        NodeId nodeId = monitoredItem.getReadValueId().getNodeId();
        Variant value = dataValue.getValue();
        DateTime sourceTime = dataValue.getSourceTime();
        log.info("节点值变化 | index={} | Identifier={} | Value={} | Time={}",
                nodeId.getNamespaceIndex().intValue(),
                nodeId.getIdentifier(),
                value != null ? value.getValue() : "",
                sourceTime != null ? sourceTime.getJavaTime() : System.currentTimeMillis());

    }


    public static void main(String[] args) throws Exception {
        // 创建 OpcUaClient（假设已连接）
        OpcUaClient client = OpcUaClientInstance.getInstance().getOpcUaClient("opc.tcp://192.168.40.103:49320", "None", null, null);
        client.connect().get();
        List<ReadNodeEntity> nodes = Arrays.asList(
                ReadNodeEntity.builder().index(2).identifier("模拟器示例._Statistics._TxBytes").build()
        );

        KepwareSubscriptManager kepwareSubscriptManager = KepwareSubscriptManager.getInstance();
        kepwareSubscriptManager.subscribeNodes(nodes);

    }

}
